package net.bwie.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.common.utils.DateTimeUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class AdjustIsNewProcessFunction extends KeyedProcessFunction<String,String,String> {

    private transient ValueState<String> firstDateState;

    @Override
    public void open(Configuration parameters) throws Exception {
        firstDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("firstDate",String.class));
    }
    @Override
    public void processElement(String s, Context context, Collector<String> collector) throws Exception {
        String firstDate = firstDateState.value();
        JSONObject jsonObject = JSON.parseObject(s);
        String is_new = jsonObject.getJSONObject("common").getString("is_new");
        String date = DateTimeUtil.convertLongToString(jsonObject.getLong("ts"), DateTimeUtil.DATE_FORMAT);
        // 状态为新用户
        if ("1".equals(is_new)) {
            // 首次登录日期为null
            if (firstDate == null) {
                // 状态正确，更新首次登录日期
                firstDateState.update(date);
                // 首次登录日期不为null
            } else {
                // 状态错误，更新状态
                if (!date.equals(firstDate)) {
                    jsonObject.getJSONObject("common").put("is_new", "0");
                }
            }
            // 状态为老用户
        } else {
            // 首次登录日期为null
            if (firstDate == null) {
                // 把当前日期前一天当做首次登录日期，更新日期
                date = DateTimeUtil.convertLongToString(jsonObject.getLong("ts") - 86400000, DateTimeUtil.DATE_FORMAT);
                firstDateState.update(date);
            }
        }
        collector.collect(jsonObject.toJSONString());
    }
}